﻿// Copyright (c) 2012 - 2021 Jared Zheng <jaredz at outlook dot com>
// Buddha Framework is licensed under Mulan PSL v2.
// You can use this software according to the terms and conditions of the Mulan PSL v2.
// You may obtain a copy of Mulan PSL v2 at:
//          http://license.coscl.org.cn/MulanPSL2
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
// See the Mulan PSL v2 for more details.
//
// Create : 2012-12-01 version 0.1 Jared Zheng <jaredz at outlook dot com>
// Update : 2021-03-19 version 4.0.0 Jared Zheng <jaredz at outlook dot com>

#include "stdafx.h"
#include "MasterServer.h"

///////////////////////////////////////////////////////////////////////////////////////////////////////////
// CMasterServer
#define PAK_HEAD( ename, uevent ) CPAKHead ename((UEVENT)uevent, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId)
#define PAK_DATA( ename, uevent, udata ) CPAKHead ename((UEVENT)uevent, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId, (UDATA)udata)

#define NET_HEAD( ename, uevent ) CNETHead ename((UEVENT)uevent, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId)
#define NET_DATA( ename, uevent, udata ) CNETHead ename((UEVENT)uevent, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId, (UDATA)udata)

#define SYNCACK_HEAD( ename ) CNETSync ename((UEVENT)PAKE_SYNCACK, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId)
#define SYNCACK_DATA( ename, udata ) CNETSync ename((UEVENT)PAKE_SYNCACK, (URANK)COMMONR_MASTER, (USID)m_SrvData.usId, (UDATA)udata)

UInt CMasterServer::Command(PCXStr, uintptr_t)
{
    return (UInt)RET_OKAY;
}

UInt CMasterServer::Update(void)
{
    if (m_SrvData.nStatus < SERVERS_ERROR)
    {
        // network fail : error
        if (m_NetworkPtr->Check(m_pConfig->GetUpdateTick() * COMMONC_CHECK))
        {
            m_SrvData.SetMayDay(); // 网络工作线程异常
            m_SrvData.nStatus = SERVERS_ERROR;
            LOG_ERROR(m_FileLog, TF("[Master]网络线程异常, 进入异常状态"));
            m_FileLog.Flush();
            return RET_FAIL;
        }
        if (Events.GetEventQueuePtr()->CheckThreads(m_SrvData.uQCheck))
        {
            m_SrvData.SetMayDay(); // 事件队列线程异常
            m_SrvData.nStatus = SERVERS_ERROR;
            LOG_ERROR(m_FileLog, TF("[Master]事件队列线程异常, 进入异常状态"));
            m_FileLog.Flush();
            return RET_FAIL;
        }
        m_RoutinePtr->Update();
        // tcp listen : warn
        CheckListen();
        CheckUpdate();
        SyncUpdate();
        return RET_OKAY;
    }
    return RET_FAIL;
}

void CMasterServer::CheckListen(void)
{
    if (m_SrvData.sListen == 0)
    {
        if (StartListen() && m_SrvData.IsWarnLoad())
        {
            LOG_WARN(m_FileLog, TF("[Master]监听重开, 尝试消除告警状态"));
            m_FileLog.Flush();
            m_SrvData.Clear();
            if (m_SrvData.IsWarnLoad(false) == false)
            {
                m_SrvData.usLoad += m_pConfig->GetServerLoad(COMMONR_MASTER);
            }
        }
    }
}

void CMasterServer::CheckUpdate(void)
{
    if (CEventQueue::CheckTickEvent(CHECK_TIME_ID, COMMONC_CHECK))
    {
        LOG_WARN(m_FileLog, TF("[Master]超时检测线程异常, 重启定时器"));
        m_FileLog.Flush();
        CEventQueue::DestroyTickEvent(CHECK_TIME_ID);
        CEventQueue::CreateTickEvent(CHECK_TIME_ID, m_pConfig->GetUpdateTick() * TIMET_S2MS, *this);
    }
    if (m_SrvData.usUpdate == (UShort)SERVERU_SYNC)
    {
        LOG_INFO(m_FileLog, TF("[Master]状态更新, 通知管理更新数据"));
        m_SrvData.usUpdate = (UShort)SERVERU_OKAY;
        m_MgrHandler->OnHandle(PAKE_UPDATEACK, reinterpret_cast<uintptr_t>(&m_SrvData), COMMONR_MASTER);
    }
}

void CMasterServer::SyncUpdate(void)
{
    for (UInt i = COMMONR_MIN; i < COMMONR_MAX; ++i)
    {
        SyncServer(i);
    }
}

void CMasterServer::SyncServer(UInt uRank)
{
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    if (Cluster.Status.usUpdate == (UShort)SERVERU_SYNC)
    {
        ARY_SYNC aSocket;
        bool bRet = m_Sync[uRank].GetSync(aSocket);
        {
            CSyncLockWaitScope scope(Cluster.Lock);
            if (bRet)
            {
                PINDEX index = Cluster.Info.GetFirstIndex();
                while (index != nullptr)
                {
                    bRet = SyncServerPack(Cluster, index, aSocket, uRank);
                }
                LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息完成"), uRank, aSocket.GetSize());
            }
            else
            {
                bRet = true;
                LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态更新, 没有订阅的服务, 直接更新为完成状态"), uRank);
            }
            if (bRet)
            {
                Cluster.Update();
                Cluster.Status.usUpdate = (UShort)SERVERU_OKAY;
            }
        }
        if (bRet)
        {
            LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态更新, 通知管理更新数据"), uRank);
            m_MgrHandler->OnHandle(PAKE_SYNC, reinterpret_cast<uintptr_t>(&Cluster.Status), uRank);
        }
    }
}

bool CMasterServer::SyncServerPack(SERVER_CLUSTER& Cluster, PINDEX& index, ARY_SYNC& aSocket, UInt uRank)
{
    SYNCACK_DATA(SyncAck, uRank);
    if (SyncAck.Alloc(m_NetworkPtr, true))
    {
        CStream& Stream = SyncAck.GetStream(false);

        UInt uCount = 0;
        while (Cluster.CheckWrite(uCount, index, Stream));

        if (uCount > 0)
        {
            SyncAck.SetStatus(Cluster.Status);
            SERVER_STATUS& Status = SyncAck.GetStatus();
            Status.usRate   = (UShort)uCount; // 当前同步的数量
            Status.usUpdate = (index != nullptr) ? (UShort)SERVERU_NONE : (UShort)SERVERU_SYNC;

            LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的服务(数量=%d)同步%d服务信息"), uRank, aSocket.GetSize(), uCount);
            CNetStreamPtr StreamPtr;
            SyncAck.GetBuffer(m_NetworkPtr, StreamPtr);
            m_NetworkPtr->Send(aSocket, StreamPtr);
        }
        else
        {
            LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息为空"), uRank, aSocket.GetSize());
        }
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息申请缓存失败"), uRank, aSocket.GetSize());
        m_FileLog.Flush();
    }
    return true;
}
//--------------------------------------
UInt CMasterServer::Init(CEventHandler& EventHandlerRef, CServerConfig& ConfigRef, UInt uRank)
{
    if (m_SrvData.nStatus == SERVERS_NONE)
    {
        // 创建服务日志
        if (m_FileLog.IsValid() == false)
        {
            m_FileLog.SetLevel(ConfigRef.GetFrameLogLevel());
            m_FileLog.Create(ConfigRef.GetLogPath(), ConfigRef.GetServerName(COMMONR_MASTER));
        }
        LOG_INFO(m_FileLog, TF("[Master]========================================"));
        LOG_INFO(m_FileLog, TF("[Master]初始化处理开始!"));
        m_MgrHandler    = &EventHandlerRef;
        m_pConfig       = &ConfigRef;
        m_SrvData.uRank = uRank;
        assert(m_SrvData.uRank == COMMONR_MASTER);
        // 从配置加载数据
        if (LoadConfig() == false)
        {
            return (UInt)RET_FAIL;
        }
        // 按配置信息加载业务层
        if (LoadRoutine() == false)
        {
            return (UInt)RET_FAIL;
        }
        m_SrvData.nStatus = SERVERS_INIT;
        LOG_INFO(m_FileLog, TF("[Master]初始化处理完成!"));
        return (UInt)RET_OKAY;
    }
    return (UInt)RET_FAIL;
}

bool CMasterServer::LoadConfig(void)
{
    // 配置信息输出到日志
    assert(m_pConfig != nullptr);
    m_pConfig->Dump(m_FileLog, COMMONR_MASTER);
    // 从动态变量表获取网络层智能指针
    assert(m_NetworkPtr == nullptr);
    m_NetworkPtr = m_pConfig->GetKeyValue().GetPtrValue<CNetwork>(CServerConfig::NetworkPtr);
    if (m_NetworkPtr == nullptr)
    {
        LOG_ERROR(m_FileLog, TF("[Master]从动态变量表获取网络层智能指针无效"));
        m_FileLog.Flush();
        return false;
    }
    // 保存Master指针到动态变量表
    m_pConfig->GetKeyValue().SetValue(CServerConfig::MasterServer, this);

    // 从配置加载数据
    assert(m_SrvData.sListen == 0);
    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(COMMONR_MASTER);
    m_SrvData.usId   = sc.usId;
    m_SrvData.usLoad = sc.usLoad;
    m_SrvData.usRate = sc.usRate;
    m_SrvData.SetUpdate(); // 更新MgrHandler信息

    m_SrvData.uQCheck = (UInt)(m_pConfig->GetUpdateTick() * COMMONC_CHECK);
    return true;
}

bool CMasterServer::LoadRoutine(void)
{
    assert(m_RoutinePtr == nullptr);

    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(COMMONR_MASTER);
    if (CObjectTraits::Load<ICommonRoutine>(sc.strInst, sc.strModule.GetBuffer(), m_RoutinePtr, sc.strRoutine.GetBuffer()) == false)
    {
        LOGV_ERROR(m_FileLog, TF("[Master]加载业务层Routine=%s, Inst=%s, Module=%s失败"),
                   sc.strRoutine.GetBuffer(), sc.strInst.GetBuffer(), sc.strModule.GetBuffer());
        m_FileLog.Flush();
        return false;
    }
    LOGV_INFO(m_FileLog, TF("[Master]加载业务层Routine=%s, Inst=%s, Module=%s完成"),
              sc.strRoutine.GetBuffer(), sc.strInst.GetBuffer(), sc.strModule.GetBuffer());
    if (m_RoutinePtr->Load(this, COMMONR_MASTER))
    {
        // 加载服务信息m_SrvData扩展数据
        // uSync=(1 << COMMON_RANK_ENUM)的组合, 表示同步其他服务的标志位
        // StrAddr, strRoutine额外的地址信息和业务附加数据
        Events.SetThreads((Int)m_pConfig->GetServerQueue(COMMONR_MASTER));
        Events.SetThreadEvent(PAKE_QUEUESTART, PAKE_QUEUESTOP);
        return true;
    }
    return false;
}
//--------------------------------------
void CMasterServer::Exit(void)
{
    if (m_SrvData.nStatus != SERVERS_NONE)
    {
        m_SrvData.nStatus = SERVERS_NONE;

        LOG_INFO(m_FileLog, TF("[Master]结束处理开始!"));
        UnloadRoutine();
        UnloadConfig();

        m_pConfig    = nullptr;
        m_MgrHandler = nullptr;
        LOG_INFO(m_FileLog, TF("[Master]结束处理完成!"));
    }
}

void CMasterServer::UnloadConfig(void)
{
    m_SrvData.Reset();
    m_SrvData.ResetData();

    m_pConfig->GetKeyValue().Remove(CServerConfig::MasterServer);
    m_NetworkPtr = nullptr;
}

void CMasterServer::UnloadRoutine(void)
{
    Events.Exit();
    if (m_RoutinePtr != nullptr)
    {
        m_RoutinePtr->Unload();
        m_RoutinePtr = nullptr;
    }
}
//--------------------------------------
bool CMasterServer::Start(void)
{
    if (m_SrvData.nStatus == SERVERS_INIT)
    {
        LOG_INFO(m_FileLog, TF("[Master]开启处理开始!"));
        Events.Init();
        if (CObjectTraits::Connect(Events, *m_RoutinePtr) == nullptr)
        {
            return false;
        }
        if (Events.StartQueue() == false)
        {
            return false;
        }
        if (m_RoutinePtr->Start() == false)
        {
            return false;
        }
        if (StartListen() == false)
        {
            return false;
        }
        // 开启检测更新线程异常的线程
        LOGV_INFO(m_FileLog, TF("[Master]开启检测更新线程异常的定时器id=%d!"), CHECK_TIME_ID);
        CEventQueue::CreateTickEvent(CHECK_TIME_ID, m_pConfig->GetUpdateTick() * TIMET_S2MS, *this);

        m_SrvData.nStatus = SERVERS_RUN;
        LOG_INFO(m_FileLog, TF("[Master]开启处理完成!"));
    }
    return true;
}

bool CMasterServer::StartListen(void)
{
    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(COMMONR_MASTER);
    if (m_SrvData.sListen != 0)
    {
        LOGV_INFO(m_FileLog, TF("[Master]监听%s:%d地址的Socket已经存在"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort);
        return true;
    }
    if (m_pConfig->GetAnyListen() != 0)
    {
        m_SrvData.sListen = m_NetworkPtr->Create(*this, sc.Listen.usPort);
    }
    else
    {
        m_SrvData.sListen = m_NetworkPtr->Create(*this, sc.Listen.usPort, sc.Listen.strIp.GetBuffer());
    }
    bool bRet = false;
    if (m_SrvData.sListen != 0)
    {
        bRet = m_NetworkPtr->Listen(m_SrvData.sListen);
        LOGV_INFO(m_FileLog, TF("[Master]监听%s:%d%s"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort, bRet ? TF("成功") : TF("失败"));
        if (bRet)
        {
            Events(PAKE_TCPLISTEN, (uintptr_t)m_SrvData.sListen, COMMONR_MASTER);
            m_MgrHandler->OnHandle(PAKE_TCPLISTEN, (uintptr_t)m_SrvData.sListen, COMMONR_MASTER);
        }
    }
    else
    {
        LOGV_ERROR(m_FileLog, TF("[Master]创建%s:%d的Socket失败"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort);
        m_FileLog.Flush();
    }
    return bRet;
}
//--------------------------------------
void CMasterServer::Stop(bool bUnlink)
{
    if (m_SrvData.nStatus > SERVERS_INIT)
    {
        if (bUnlink)
        {
            CAtomics::Exchange<UInt>((PUInt)&m_SrvData.nStatus, SERVERS_UNLINK);
            m_RoutinePtr->Stop();
            Events.Exit();
        }
        else
        {
            LOG_INFO(m_FileLog, TF("[Master]停止处理开始!"));
            LOGV_INFO(m_FileLog, TF("[Master]停止检测更新线程异常的定时器id=%d!"), CHECK_TIME_ID);
            CEventQueue::DestroyTickEvent(CHECK_TIME_ID);

            CAtomics::Exchange<UInt>((PUInt)&m_SrvData.nStatus, SERVERS_STOP);
            StopListen();

            for (UInt i = COMMONR_MIN; i < COMMONR_MAX; ++i)
            {
                m_Cluster[i].Info.RemoveAll();
                m_Cluster[i].Reset();

                m_Sync[i].Reset();
            }

            m_SrvData.Zero();
            m_SrvData.usLoad = m_pConfig->GetServerLoad(COMMONR_MASTER);

            m_SrvData.nStatus = SERVERS_INIT;
            LOG_INFO(m_FileLog, TF("[Master]停止处理完成!"));
        }
    }
}

void CMasterServer::StopListen(void)
{
    if (m_SrvData.sListen != 0)
    {
        m_NetworkPtr->Destroy(m_SrvData.sListen);
        m_SrvData.sListen = 0;
        LOG_INFO(m_FileLog, TF("[Master]停止监听Socket"));
    }
}
//--------------------------------------
bool CMasterServer::OnTcpDispatch(const PacketPtr& PktPtr, PTCP_PARAM pTcp)
{
    // 1. PTCP_PARAM->pCache != nullptr时上层可以保存网络层申请的缓存数据
    //    PTCP_PARAM->pCache, PTCP_PARAM->index是网络层申请的缓存数据的起始地址和索引号
    // 2. PTCP_PARAM->pData为当前数据开始地址, PTCP_PARAM->stSize为当前数据大小
    // 3. 如果上层保存缓存, 设置PTCP_PARAM->pData为nullptr通知网络层不要再使用当前缓存
    // 4. 释放缓存使用MObject::sMCFree(index, pCache)
    bool bRet  = true;
    UInt uRank = PktPtr->GetRank();
    if ((uRank > COMMONR_MASTER) && (uRank < COMMONR_MAX))
    {
        switch (PktPtr->GetEvent())
        {
        case PAKE_LIVE:
            {
                // 统计在线服务数量, 心跳时将在线数量回复给对应的服务
                SERVER_STATUS& ss = m_Cluster[uRank].Status;
                PAK_DATA(LiveAck, PAKE_LIVEACK, ss.uOnline);
                m_NetworkPtr->Send(pTcp->sSocket, LiveAck);
            }
            break;
        case PAKE_LIVEACK:
        case PAKE_LINKACK:
        case PAKE_UPDATEACK:
        case PAKE_SYNCACK:
        case PAKE_UNLINKACK:
            {
                // 忽略不处理的通用信令
            }
            break;
        case PAKE_LINK:
        case PAKE_UPDATE:
        case PAKE_SYNC:
        case PAKE_UNLINK:
            {
                // 需要处理的通用信令, 并且需要传递给业务层
                bRet = OnCommonEvent(uRank, PktPtr, pTcp->sSocket);
                if (bRet == false)
                {
                    break;
                }
            }
            //break;
        default:
            {
                if (PktPtr->IsStream())
                {
                    bRet = PktPtr->Bind(m_NetworkPtr, pTcp);
                }
                if (bRet)
                {
                    Events(PktPtr->GetEvent(), *PktPtr, (ULLong)pTcp->sSocket);
                }
            }
        }
    }
    else
    {
        bRet = false;
        LOGV_WARN(m_FileLog, TF("[Master]无法识别的层级编号=%d, 信令=%d, 信令包数据大小=%d, 链接=%p"),
                  uRank, PktPtr->GetEvent(), pTcp->stSize, pTcp->sSocket);
        m_FileLog.Flush();
    }
    return bRet;
}

bool CMasterServer::OnCommonEvent(UInt uRank, const PacketPtr& PktPtr, Socket_t sSocket)
{
    bool bRet = true;
    switch (PktPtr->GetEvent())
    {
    case PAKE_LINK:
        {
            CNETHead* pLink = static_cast<CNETHead*>(PktPtr.Get());
            bRet = OnLink(uRank, sSocket, pLink);
        }
        break;
    case PAKE_UPDATE:
        {
            CNETHead* pUpdate = static_cast<CNETHead*>(PktPtr.Get());
            bRet = OnUpdate(uRank, sSocket, pUpdate);
        }
        break;
    case PAKE_SYNC:
        {
            CPAKHead* pSync = static_cast<CPAKHead*>(PktPtr.Get());
            bRet = OnSync(uRank, sSocket, pSync);
        }
        break;
    case PAKE_UNLINK:
        {
            CPAKHead* pHead = static_cast<CPAKHead*>(PktPtr.Get());
            bRet = OnUnlink(uRank, sSocket, pHead);
        }
        break;
    default:{}
    }
    return bRet;
}

bool CMasterServer::OnLink(UInt uRank, Socket_t sSocket, CNETHead* pLink)
{
    bool bRet = LinkServer(uRank, sSocket, pLink);
    if (bRet)
    {
        ULLong ullLiveData = pLink->GetData();
        ullLiveData <<= 32;   // type[high-32, master-rank(0)-16 -- link-type-16]
        ullLiveData += uRank; // rank[low-32]

        m_NetworkPtr->SetAttr(sSocket, ullLiveData);

        // 统计在线服务数量, 将在线数量回复给对应的服务
        SERVER_STATUS& ss = m_Cluster[uRank].Status;
        PAK_DATA(LinkAck, PAKE_LINKACK, ss.uOnline);
        bRet = m_NetworkPtr->Send(sSocket, LinkAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CMasterServer::OnUpdate(UInt uRank, Socket_t sSocket, CNETHead* pUpdate)
{
    bool bRet = UpdateServer(uRank, sSocket, pUpdate);
    if (bRet)
    {
        // 统计在线服务数量, 将在线数量回复给对应的服务
        SERVER_STATUS& ss = m_Cluster[uRank].Status;
        PAK_DATA(UpdateAck, PAKE_UPDATEACK, ss.uOnline);
        bRet = m_NetworkPtr->Send(sSocket, UpdateAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CMasterServer::OnSync(UInt uRank, Socket_t sSocket, CPAKHead* pSync)
{
    UInt uComb = pSync->GetData(); // 包含需要同步的加载类型组合
    if (uComb != 0)
    {
        LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)要求同步的层级比特掩码%x"), uRank, uComb);
        for (UInt i = COMMONR_MIN; i < uRank; ++i)
        {
            UInt uSync = (1 << i);
            if ((uComb & uSync) != 0)
            {
                SyncServer(i, sSocket, pSync);
            }
        }
    }
    return true;
}

bool CMasterServer::OnUnlink(UInt uRank, Socket_t sSocket, CPAKHead* pHead)
{
    bool bRet = UnlinkServer(uRank, sSocket, pHead->GetData());
    if (bRet)
    {
        m_NetworkPtr->SetAttr(sSocket, 0);
        PAK_HEAD(UnlinkAck, PAKE_UNLINKACK);
        bRet = m_NetworkPtr->Send(sSocket, UnlinkAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CMasterServer::LinkServer(UInt uRank, Socket_t sSocket, CNETHead* pLink)
{
    assert(pLink->IsTempStream());
    CStream& Stream = pLink->GetTempStream();

    PAIR_INFO info(sSocket);
    info.m_V.Status(Stream);
    assert(info.m_V.usUpdate == (UShort)SERVERU_LINK);
    info.m_V.Info(Stream);
    LOGV_INFO(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p注册, StrAddr = %s[%d], Routine=%s, Instance=%s"), pLink->GetData(), uRank, sSocket,
              info.m_V.StrAddr.strIp.GetBuffer(), info.m_V.StrAddr.usPort, info.m_V.strRoutine.GetBuffer(), info.m_V.strInstance.GetBuffer());

    UInt uRet = Events(PAKE_LINK, (uintptr_t)info.m_V.usId, uRank, nullptr, CEventQueue::EVENT_TYPE_SYNC);
    if (uRet != RET_OKAY)
    {
        LOGV_WARN(m_FileLog, TF("[Master]服务(RANK=%d)%p注册时业务实例检测id=%d返回错误=%d"), uRank, sSocket, info.m_V.usId, uRet);
        m_FileLog.Flush();
        return false;
    }

    PINDEX index = nullptr;
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if (pPair == nullptr)
        {
            LOGV_INFO(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p注册, Id=%d, Sync-Mask=%x"), pLink->GetData(), uRank, sSocket, info.m_V.usId, info.m_V.uSync);
            index = Cluster.Info.Add(sSocket, info.m_V);
        }
        else if (pPair->m_V.usUpdate == (UShort)SERVERU_UNLINK)
        {
            // 注销的信息未同步到其他服务
            LOGV_INFO(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p重新注册, Id=%d, Sync-Mask=%x"), pLink->GetData(), uRank, sSocket, info.m_V.usId, info.m_V.uSync);
            pPair->m_V = info.m_V;
            index = reinterpret_cast<PINDEX>(pPair);
        }
        else
        {
            info.m_V.usUpdate = pPair->m_V.usUpdate;
        }
    }
    if (index != nullptr)
    {
        Cluster.Link(info.m_V);
        m_SrvData.IncrOnline();

        AddSync(uRank, info.m_V.uSync, sSocket);

        m_MgrHandler->OnHandle(PAKE_LINK, reinterpret_cast<uintptr_t>(&info), uRank);
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Master]服务(RANK=%d)%p注册时存在异常状态=%d"), uRank, sSocket, info.m_V.usUpdate);
    m_FileLog.Flush();
    return false;
}

bool CMasterServer::AddSync(UInt uRank, UInt uSync, Socket_t sSocket)
{
    if (uSync == 0)
    {
        return false;
    }
    for (UInt i = COMMONR_MIN; i < uRank; ++i)
    {
        if ((uSync & (1 << i)) != 0)
        {
            m_Sync[i].AddSync(sSocket);
        }
    }
    return true;
}

bool CMasterServer::UpdateServer(UInt uRank, Socket_t sSocket, CNETHead* pUpdate)
{
    assert(pUpdate->IsTempStream());
    CStream& Stream = pUpdate->GetTempStream();

    PAIR_INFO info(sSocket);
    info.m_V.Status(Stream);
    assert(info.m_V.usUpdate == (UShort)SERVERU_SYNC);

    SERVER_STATUS ss;
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if ( (pPair != nullptr) &&
             (pPair->m_V.usUpdate > (UShort)SERVERU_NONE) &&
             (pPair->m_V.usUpdate < (UShort)SERVERU_UNLINK) )
        {
            ss = pPair->m_V;
            pPair->m_V.Copy(info.m_V);
        }
        else if (pPair != nullptr)
        {
            ss.usLoad = pPair->m_V.usUpdate;
        }
    }
    if (ss.usUpdate > (UShort)SERVERU_NONE)
    {
        Cluster.Update(info.m_V, ss);
        m_MgrHandler->OnHandle(PAKE_UPDATE, reinterpret_cast<uintptr_t>(&info), uRank);
        return true;
    }
    if (ss.usLoad != (UShort)SERVERU_NONE)
    {
        LOGV_WARN(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p更新时存在异常状态=%d"), pUpdate->GetData(), uRank, sSocket, ss.usLoad);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p更新时不存在"), pUpdate->GetData(), uRank, sSocket);
    }
    m_FileLog.Flush();
    return false;
}

bool CMasterServer::SyncServer(UInt uRank, Socket_t sSocket, CPAKHead* pSync)
{
    LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)要求同步的层级编号%x"), pSync->GetRank(), uRank);
    bool bRet = true;
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PINDEX index = Cluster.Info.GetFirstIndex();
        do { bRet = SyncServerPack(Cluster, index, sSocket, uRank); } while (index != nullptr);
    }
    return bRet;
}

bool CMasterServer::SyncServerPack(SERVER_CLUSTER& Cluster, PINDEX& index, Socket_t sSocket, UInt uRank)
{
    SYNCACK_DATA(SyncAck, uRank);
    if (SyncAck.Alloc(m_NetworkPtr, true))
    {
        CStream& Stream = SyncAck.GetStream(false);

        UInt uCount = 0;
        while (Cluster.CheckWrite(uCount, index, Stream, (UShort)SERVERU_OKAY_ALL));

        SyncAck.SetStatus(Cluster.Status);
        SERVER_STATUS& Status = SyncAck.GetStatus();
        Status.usRate   = (UShort)uCount; // 当前同步的数量
        Status.usUpdate = (index != nullptr) ? (UShort)SERVERU_NONE : (UShort)SERVERU_SYNC;

        LOGV_INFO(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的单个服务(socket=%p)同步%d服务信息"), uRank, sSocket, uCount);
        CNetStreamPtr StreamPtr;
        SyncAck.GetBuffer(m_NetworkPtr, StreamPtr);
        m_NetworkPtr->Send(sSocket, StreamPtr);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Master](RANK=%d)的集群状态向订阅的单个服务(socket=%p)同步服务信息申请缓存失败"), uRank, sSocket);
        m_FileLog.Flush();
    }
    return true;
}

bool CMasterServer::UnlinkServer(UInt uRank, Socket_t sSocket, UInt uType)
{
    PAIR_INFO info(sSocket);
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if (pPair != nullptr)
        {
            info.m_V = pPair->m_V;
            if (info.m_V.usUpdate == (UShort)SERVERU_LINK)
            {
                // 服务从注册到注销, 信息还未同步到其他服务, 直接作为无状态信息删除
                Cluster.Info.RemoveAt(reinterpret_cast<PINDEX>(pPair));
            }
            else if (info.m_V.usUpdate != (UShort)SERVERU_UNLINK)
            {
                pPair->m_V.usUpdate = (UShort)SERVERU_UNLINK; // 需要同步给其他服务以后再删除
            }
        }
    }
    if (info.m_V.usUpdate != (UShort)SERVERU_NONE)
    {
        Cluster.Unlink(info.m_V);
        m_SrvData.DecrOnline();

        RemoveSync(uRank, info.m_V.uSync, sSocket);

        Events(PAKE_UNLINK, (uintptr_t)info.m_V.usId, uRank, nullptr, CEventQueue::EVENT_TYPE_SYNC);

        m_MgrHandler->OnHandle(PAKE_UNLINK, reinterpret_cast<uintptr_t>(&info), uRank);
        LOGV_INFO(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p注销, Id=%d, Sync-Mask=%x"), uType, uRank, sSocket, info.m_V.usId, info.m_V.uSync);
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Master]服务(TYPE=%d, RANK=%d)%p注销时不存在"), uType, uRank, sSocket);
    m_FileLog.Flush();
    return false;
}

bool CMasterServer::RemoveSync(UInt uRank, UInt uSync, Socket_t sSocket)
{
    if (uSync == 0)
    {
        return false;
    }
    for (UInt i = COMMONR_MIN; i < uRank; ++i)
    {
        if ((uSync & (1 << i)) != 0)
        {
            m_Sync[i].RemoveSync(sSocket);
        }
    }
    return true;
}
//--------------------------------------
bool CMasterServer::OnTcpAccept(Socket_t sAccept, Socket_t)// sListen)
{
    if ((m_SrvData.nStatus == SERVERS_RUN) && (m_SrvData.IsOverLoad(false) == false))
    {
        CStrAddr strAddr;
        m_NetworkPtr->GetAddr(sAccept, strAddr.strIp, strAddr.usPort);
        LOGV_DUMP(m_FileLog, TF("[Master]%s:%d连接完成!"), strAddr.strIp.GetBuffer(), strAddr.usPort);

        m_SrvData.IncrTotal();
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Master]非运行状态STATUS=%d或者LOAD=%d!"), m_SrvData.nStatus, m_SrvData.usLoad);
    return false;
}
//--------------------------------------
bool CMasterServer::OnTcpClose(Socket_t sSocket, ULLong ullLiveData)
{
    if (CAtomics::CompareExchange<UInt>((PUInt)&m_SrvData.nStatus, 0, 0) == SERVERS_STOP)
    {
        return false;
    }
    if (sSocket == m_SrvData.sListen)
    {
        m_SrvData.sListen = 0;
        LOG_WARN(m_FileLog, TF("[Master]监听关闭, 进入告警状态"));
        m_FileLog.Flush();

        if (m_SrvData.nStatus < SERVERS_ERROR)
        {
            m_SrvData.SetWarnLoad();
        }
        Events(PAKE_TCPLISTEN_CLOSE, (uintptr_t)sSocket, COMMONR_MASTER);
        m_MgrHandler->OnHandle(PAKE_TCPLISTEN_CLOSE, (uintptr_t)sSocket, COMMONR_MASTER);
    }
    else
    { // 单个服务断开
        if (ullLiveData != 0)
        {
            UnlinkServer((UInt)ullLiveData, sSocket, (UInt)(ullLiveData >> 32));
            Events(PAKE_TCPACCEPT_CLOSE, (uintptr_t)sSocket, ullLiveData);
        }
        m_SrvData.DecrTotal();
    }
    return true;
}
//--------------------------------------
UInt CMasterServer::OnHandle(uintptr_t, UInt)
{
    if (m_SrvData.IsMayDay() == false)
    {
        // 检测更新线程定时器是否工作正常
        if (CEventQueue::CheckTickEvent(CServerConfig::TIMED_UPDATE_ID, COMMONC_CHECK))
        {
            m_SrvData.SetMayDay(); // 更新线程异常
            m_SrvData.nStatus = SERVERS_ERROR;
            LOG_ERROR(m_FileLog, TF("[Master]更新线程异常, 进入异常状态"));
            m_FileLog.Flush();
        }
    }
    if (m_SrvData.IsMayDay()) // network/queue/update thread exception
    {
        DEV_ERROR(TF("[Master]异常状态-----异常处理"));
    }
    else if (m_SrvData.IsWarnLoad()) // listen tcp close 或者其他
    {
        DEV_WARN(TF("[Master]告警状态-----告警处理"));
    }
    return RET_OKAY;
}
